"""验证会议数据集格式的独立脚本"""
import pandas as pd
import json
import sys
from pathlib import Path


def verify_parquet_file(parquet_file):
    """
    验证parquet文件的格式
    
    Args:
        parquet_file: parquet文件路径
    """
    print("=" * 80)
    print(f"验证数据文件: {parquet_file}")
    print("=" * 80)
    
    # 1. 检查文件是否存在
    parquet_path = Path(parquet_file)
    if not parquet_path.exists():
        print(f"❌ 错误: 文件不存在: {parquet_file}")
        return False
    
    # 2. 加载数据
    try:
        df = pd.read_parquet(parquet_file)
        print(f"✅ 成功加载数据文件")
        print(f"   数据量: {len(df)} 条")
        print(f"   列数: {len(df.columns)}")
    except Exception as e:
        print(f"❌ 错误: 无法加载文件: {e}")
        return False
    
    # 3. 列出所有列
    print("\n" + "=" * 80)
    print("数据列列表:")
    print("=" * 80)
    for i, col in enumerate(df.columns, 1):
        print(f"  {i}. {col}")
    
    # 4. 检查必需字段
    required_fields = [
        'task_id',
        'meeting_topic',
        'required_attendees',
        'optional_attendees',
        'availability',
        'historical_pattern',
        'agent_name',        # Agent Loop 选择器（关键！）
        'reward_model',
    ]
    
    print("\n" + "=" * 80)
    print("必需字段检查:")
    print("=" * 80)
    
    all_fields_ok = True
    for field in required_fields:
        if field in df.columns:
            print(f"✅ {field:25s}")
        else:
            print(f"❌ {field:25s} [缺失]")
            all_fields_ok = False
    
    if not all_fields_ok:
        print("\n❌ 数据格式检查失败：缺少必需字段")
        return False
    
    # 5. 检查前3个样本
    print("\n" + "=" * 80)
    print("样本数据检查 (前3条):")
    print("=" * 80)
    
    for idx in range(min(3, len(df))):
        print(f"\n样本 {idx}:")
        row = df.iloc[idx]
        
        print(f"  task_id: {row.get('task_id', 'N/A')}")
        print(f"  meeting_topic: {row.get('meeting_topic', 'N/A')}")
        print(f"  agent_name: '{row.get('agent_name', 'N/A')}' (类型: {type(row.get('agent_name', None)).__name__})")
        
        # 检查 reward_model
        reward_model = row.get('reward_model', None)
        print(f"  reward_model: {type(reward_model).__name__}")
        
        if isinstance(reward_model, dict):
            print(f"    └─ ground_truth: '{reward_model.get('ground_truth', 'N/A')}'")
        elif isinstance(reward_model, str):
            try:
                parsed = json.loads(reward_model)
                print(f"    └─ (JSON字符串) ground_truth: '{parsed.get('ground_truth', 'N/A')}'")
            except:
                print(f"    └─ (无效的JSON字符串): {reward_model[:50]}...")
        else:
            print(f"    └─ ❌ 类型错误: {type(reward_model)}")
        
        print(f"  required_attendees: {len(row.get('required_attendees', []))} 人")
        print(f"  optional_attendees: {len(row.get('optional_attendees', []))} 人")
    
    # 6. 检查 agent_name 字段的统计（关键！）
    print("\n" + "=" * 80)
    print("agent_name 字段统计:")
    print("=" * 80)
    
    if 'agent_name' in df.columns:
        agent_name_values = df['agent_name'].tolist()
        unique_values = set(agent_name_values)
        
        print(f"唯一值数量: {len(unique_values)}")
        print(f"唯一值: {unique_values}")
        
        # 统计每个值的数量
        value_counts = {}
        for val in agent_name_values:
            key = f"'{val}'" if isinstance(val, str) else str(val)
            value_counts[key] = value_counts.get(key, 0) + 1
        
        print("\n值分布:")
        for val, count in sorted(value_counts.items(), key=lambda x: -x[1]):
            print(f"  {val}: {count} 条")
    else:
        print("⚠️  agent_name 字段不存在！")
    
    # 7. 检查是否有空值
    print("\n" + "=" * 80)
    print("空值检查:")
    print("=" * 80)
    
    for col in required_fields:
        if col in df.columns:
            null_count = df[col].isnull().sum()
            print(f"  {col:25s}: {null_count} null")
    
    # 8. 导出前3个样本为JSON
    json_output_path = parquet_path.with_suffix('.verify.json')
    sample_data = df.head(3).to_dict('records')
    
    # 转换为可序列化格式
    for item in sample_data:
        for key, val in item.items():
            if isinstance(val, (pd.Series, pd.DataFrame)):
                item[key] = val.to_dict()
    
    try:
        with open(json_output_path, 'w', encoding='utf-8') as f:
            json.dump(sample_data, f, ensure_ascii=False, indent=2, default=str)
        print(f"\n✅ 已导出前3个样本到: {json_output_path}")
    except Exception as e:
        print(f"\n⚠️  导出JSON失败: {e}")
    
    print("\n" + "=" * 80)
    print("验证完成！")
    print("=" * 80)
    
    return True


def main():
    if len(sys.argv) < 2:
        print("用法: python verify_data.py <parquet文件路径>")
        print("\n示例:")
        print("  python verify_data.py /path/to/train.parquet")
        print("  python verify_data.py /path/to/test.parquet")
        sys.exit(1)
    
    parquet_file = sys.argv[1]
    success = verify_parquet_file(parquet_file)
    
    sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()

